package core.SparkStreaming;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.streaming.Duration;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaInputDStream;
import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.spark.streaming.kafka010.ConsumerStrategies;
import org.apache.spark.streaming.kafka010.KafkaUtils;
import org.apache.spark.streaming.kafka010.LocationStrategies;


import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;

public class Spark03_KAFKA {

    public static void main(String[] args) {

        SparkConf sparkConf = new SparkConf();
        sparkConf.setMaster("local[*]");
        sparkConf.setAppName("Spark03_KAFKA");
        try {

            JavaStreamingContext jsc = new JavaStreamingContext(sparkConf, new Duration(3 * 1000));


            HashMap<String, Object> map = new HashMap<>();

            // 1. BOOTSTRAP_SERVERS_CONFIG: 指定 Kafka 集群的地址，即 broker 的主机和端口。
            // 这里填写的是 Kafka 服务器的地址和端口号，消费者通过它连接到 Kafka 集群。
            map.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "114.251.235.19:9092");

            // 2. KEY_DESERIALIZER_CLASS_CONFIG: 指定 Key 的反序列化器类。
            // Kafka 消息的 Key 是以字节数组的形式存储的，需要通过反序列化器将其转换为 Java 对象。
            // 常用的反序列化器类是 org.apache.kafka.common.serialization.StringDeserializer（字符串反序列化）。
            map.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

            // 3. VALUE_DESERIALIZER_CLASS_CONFIG: 指定 Value 的反序列化器类。
            // Kafka 消息的 Value 同样是以字节数组的形式存储的，需要通过反序列化器将其转换为 Java 对象。
            // 常用的反序列化器类是 org.apache.kafka.common.serialization.StringDeserializer（字符串反序列化）。
            map.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

            // 4. GROUP_ID_CONFIG: 指定消费者所属的消费者组。
            // Kafka 使用消费者组来管理多个消费者的负载均衡和消息分区分配。
            // 不同的消费者组可以独立消费同一个主题的消息。
            map.put(ConsumerConfig.GROUP_ID_CONFIG, "my-consumer-group");

            // 5. AUTO_OFFSET_RESET_CONFIG: 指定当消费者没有初始偏移量或当前偏移量不存在时的行为。
            // 可选值：
            // - "earliest": 自动从最早的消息开始消费。
            // - "latest": 自动从最新的消息开始消费（默认值）。
            // - "none": 如果没有找到消费者的偏移量，则抛出异常。
            map.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");


            ArrayList<String> strings = new ArrayList<>();
            strings.add("topic_test");


            JavaInputDStream<ConsumerRecord<String, String>> stream = KafkaUtils.createDirectStream(
                    jsc,
                    LocationStrategies.PreferConsistent(),
                    ConsumerStrategies.<String, String>Subscribe(
                            strings,
                            map
                    )
            );




            JavaDStream<String> javaDStreamRdd = stream.map(new Function<ConsumerRecord<String, String>, String>() {
                @Override
                public String call(ConsumerRecord<String, String> v1) throws Exception {
                    return v1.value();
                }
            });

            javaDStreamRdd.foreachRDD(rdd -> {
                if (!rdd.isEmpty()) {
                    System.out.println("Received data: " + rdd.collect());
                } else {
                    System.out.println("No data received.");
                }
            });

//            JavaInputDStream<ConsumerRecord<String,String>>


            /**
             * 启动数据采集器
             */
            jsc.start();

            /**
             * 等待数据采集器结束，如果采集器结束运行，那么main线程就会继续执行
             */
            jsc.awaitTermination();


        } catch (Exception e) {
            e.printStackTrace(); // 捕获并打印异常
        } finally {
//            sparkSession.close();
        }
    }

}

